package com.gugee.ins.data.collector.mblogger.listener;

import com.gugee.ins.data.collector.mblogger.config.InsMBloggerConfig;
import com.gugee.ins.data.collector.mblogger.service.MBloggerSvc;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description:完成大播主的收录及更新
 * User: luolifeng
 * Date: 2020-12-24 11:13
 */

@Component
@Slf4j
public class InsMBloggerListener {

    @Autowired
    private InsMBloggerConfig insMBloggerConfig;
    @Autowired
    private MBloggerSvc mBloggerSvc;
    private final MBloggerCollectHandler collectHandler=new MBloggerCollectHandler();
    private final MBloggerUpdateHandler updateHandler=new MBloggerUpdateHandler();


    /**
     * 监听任务,完成大播主的收录
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerCl.id}",topics = {"${kafka.topics.insMBloggerCl.online}"},containerFactory = "factoryMBloggerClAck")
    public void bloggerMainCollect(List<InsBloggerMessage> messages, Acknowledgment ack) {
        collectHandler.handle("bloggerMainCollect",messages,insMBloggerConfig.getAcceptMsgClMaxLimitTime(),ack);
    }


    /**
     * 监听任务,完成大播主的更新
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerUpdate.id}",topics = {"${kafka.topics.insMBloggerUpdate.online}"},containerFactory = "factoryMBloggerUpdateAck")
    public void bloggerMainUpdate(List<InsBloggerMessage> messages, Acknowledgment ack) {
        updateHandler.handle("bloggerMainUpdate",messages,insMBloggerConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 监听任务,完成中播主的更新
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMiddleBloggerUpdate.id}",
            topics = {"${kafka.topics.insMiddleBloggerUpdate.online}"},
            containerFactory = "factoryMiddleBloggerUpdateAck")
    public void middleBloggerMainUpdate(List<InsBloggerMessage> messages, Acknowledgment ack) {
        updateHandler.handle("middleBloggerMainUpdate",messages,insMBloggerConfig.getAcceptMsgMiddleBloggerUpdateMaxLimitTime(),ack);
    }

    class MBloggerCollectHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            mBloggerSvc.handleCollect(messages);
        }
    }
    class MBloggerUpdateHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            mBloggerSvc.handleUpdate(messages);
        }
    }
}
